package com.song.storm_demo.chap_17_1;

import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.builtin.Count;
import storm.trident.testing.FixedBatchSpout;
import storm.trident.testing.MemoryMapState;
import storm.trident.testing.Split;
import storm.trident.tuple.TridentTuple;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class TridentWordCount {

	public static class Split extends BaseFunction
	{

		@Override
		public void execute(TridentTuple tuple, TridentCollector collector) {
			 String sentence = tuple.getString(0);
			 String[] seg = sentence.split(" ");
			 for(String k :seg)
			 {
				 collector.emit(new Values(k));
			 }
			
		}
		
	}
	
	public static StormTopology buildTopology(LocalDRPC drpc)
	{
		
	}
	
	
	public static  void main(String[] args)
	{
		
		FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3, 
				new Values("the cow jumped over thre moon"),
				new Values("the man went to the and"),
				new Values("four score and seven years ago"),
				new Values("how many apples can you eat")
				);
		spout.setCycle(true);
		
		TridentTopology topology = new TridentTopology();
		TridentState wordCounts = topology.newStream("spout1", spout).parallelismHint(16)
				.each(new Fields("sentence"), new Split(), new Fields("word"))
				.groupBy(new Fields("word"))
				.persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
				.parallelismHint(16);
		
		 	LocalCluster cluster = new LocalCluster();
		    Config config = new Config();
		    config.setDebug(true);
		    config.setMaxSpoutPending(3);
		    cluster.submitTopology("global-count-topology", config,topology.build());
		    try {
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		    cluster.shutdown();
		
		
	}
	
	
}
